package com.ksyun.campus.metaserver.client;

import com.ksyun.campus.metaserver.config.MinfsProperties;
import com.ksyun.campus.metaserver.domain.DataServerInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;

import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.Map;

/**
 * Client for communicating with DataServer instances
 */
@Slf4j
@Component
public class DataServerClient {
    
    @Autowired
    private MinfsProperties properties;
    
    private final WebClient webClient;
    
    public DataServerClient() {
        this.webClient = WebClient.builder()
                .codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(10 * 1024 * 1024)) // 10MB
                .build();
    }
    
    /**
     * Request data server to replicate a block
     */
    public boolean replicateBlock(DataServerInfo srcDs, DataServerInfo dstDs, 
                                String fileId, int blkIdx, long size) {
        try {
            String url = String.format("http://%s:%d/api/v1/data/replicate", dstDs.getHost(), dstDs.getPort());
            
            Map<String, Object> request = Map.of(
                    "src", Map.of(
                            "host", srcDs.getHost(),
                            "port", srcDs.getPort(),
                            "fileId", fileId,
                            "blkIdx", blkIdx
                    ),
                    "dst", Map.of(
                            "host", dstDs.getHost(),
                            "port", dstDs.getPort()
                    ),
                    "size", size
            );
            
            String response = webClient.post()
                    .uri(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(request)
                    .retrieve()
                    .bodyToMono(String.class)
                    .timeout(Duration.ofMillis(properties.getDataserver().getClient().getReadTimeout()))
                    .retryWhen(Retry.fixedDelay(
                            properties.getDataserver().getClient().getMaxRetries(),
                            Duration.ofSeconds(1)))
                    .block();
            
            log.debug("Replicate block response from {}: {}", dstDs.getDsId(), response);
            return true;
            
        } catch (Exception e) {
            log.error("Failed to replicate block {}:{} from {} to {}", 
                    fileId, blkIdx, srcDs.getDsId(), dstDs.getDsId(), e);
            return false;
        }
    }
    
    /**
     * Request data server to delete a block
     */
    public boolean deleteBlock(DataServerInfo ds, String fileId, int blkIdx) {
        try {
            String url = String.format("http://%s:%d/api/v1/data/deleteBlock", ds.getHost(), ds.getPort());
            
            Map<String, Object> request = Map.of(
                    "fileId", fileId,
                    "blkIdx", blkIdx
            );
            
            String response = webClient.post()
                    .uri(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(request)
                    .retrieve()
                    .bodyToMono(String.class)
                    .timeout(Duration.ofMillis(properties.getDataserver().getClient().getReadTimeout()))
                    .retryWhen(Retry.fixedDelay(
                            properties.getDataserver().getClient().getMaxRetries(),
                            Duration.ofSeconds(1)))
                    .block();
            
            log.debug("Delete block response from {}: {}", ds.getDsId(), response);
            return true;
            
        } catch (Exception e) {
            log.error("Failed to delete block {}:{} from {}", fileId, blkIdx, ds.getDsId(), e);
            return false;
        }
    }
    
    /**
     * Get health and capacity information from data server
     */
    public DataServerHealthInfo getHealth(DataServerInfo ds) {
        try {
            String url = String.format("http://%s:%d/api/v1/data/health", ds.getHost(), ds.getPort());
            
            @SuppressWarnings("unchecked")
            Map<String, Object> response = webClient.get()
                    .uri(url)
                    .retrieve()
                    .bodyToMono(Map.class)
                    .timeout(Duration.ofMillis(properties.getDataserver().getClient().getConnectTimeout()))
                    .retryWhen(Retry.fixedDelay(2, Duration.ofSeconds(1)))
                    .block();
            
            return DataServerHealthInfo.builder()
                    .healthy(true)
                    .capacityBytes(((Number) response.getOrDefault("capacityBytes", 0L)).longValue())
                    .usedBytes(((Number) response.getOrDefault("usedBytes", 0L)).longValue())
                    .build();
            
        } catch (WebClientResponseException e) {
            if (e.getStatusCode() == HttpStatus.SERVICE_UNAVAILABLE) {
                log.warn("DataServer {} is not ready", ds.getDsId());
            } else {
                log.error("Error getting health from DataServer {}: {}", ds.getDsId(), e.getMessage());
            }
            return DataServerHealthInfo.builder().healthy(false).build();
            
        } catch (Exception e) {
            log.error("Failed to get health from DataServer {}", ds.getDsId(), e);
            return DataServerHealthInfo.builder().healthy(false).build();
        }
    }
    
    /**
     * Check if data server is responsive
     */
    public boolean ping(DataServerInfo ds) {
        try {
            String url = String.format("http://%s:%d/api/v1/data/ping", ds.getHost(), ds.getPort());
            
            webClient.get()
                    .uri(url)
                    .retrieve()
                    .bodyToMono(String.class)
                    .timeout(Duration.ofMillis(5000))
                    .block();
            
            return true;
            
        } catch (Exception e) {
            log.debug("Ping failed for DataServer {}: {}", ds.getDsId(), e.getMessage());
            return false;
        }
    }
    
    /**
     * Check if a specific block exists on the data server
     * 
     * @param ds DataServer information
     * @param fileId File ID
     * @param blkIdx Block index
     * @return true if the block exists, false otherwise
     */
    public boolean blockExists(DataServerInfo ds, String fileId, int blkIdx) {
        try {
            String url = String.format("http://%s:%d/api/v1/data/blockExists?fileId=%s&blkIdx=%d", 
                    ds.getHost(), ds.getPort(), 
                    java.net.URLEncoder.encode(fileId, java.nio.charset.StandardCharsets.UTF_8), 
                    blkIdx);
            
            @SuppressWarnings("unchecked")
            Map<String, Object> response = webClient.get()
                    .uri(url)
                    .retrieve()
                    .bodyToMono(Map.class)
                    .timeout(Duration.ofMillis(properties.getDataserver().getClient().getReadTimeout()))
                    .retryWhen(Retry.fixedDelay(2, Duration.ofSeconds(1)))
                    .block();
            
            Boolean exists = (Boolean) response.get("exists");
            log.debug("Block exists check on {}: {}:{} -> {}", ds.getDsId(), fileId, blkIdx, exists);
            return Boolean.TRUE.equals(exists);
            
        } catch (WebClientResponseException e) {
            if (e.getStatusCode() == HttpStatus.NOT_FOUND) {
                log.debug("Block {}:{} not found on DataServer {}", fileId, blkIdx, ds.getDsId());
            } else {
                log.warn("Error checking block existence on DataServer {}: {}", ds.getDsId(), e.getMessage());
            }
            return false;
            
        } catch (Exception e) {
            log.error("Failed to check block existence {}:{} on DataServer {}", fileId, blkIdx, ds.getDsId(), e);
            return false;
        }
    }
    
    /**
     * Data server health information
     */
    public static class DataServerHealthInfo {
        private final boolean healthy;
        private final long capacityBytes;
        private final long usedBytes;
        
        private DataServerHealthInfo(boolean healthy, long capacityBytes, long usedBytes) {
            this.healthy = healthy;
            this.capacityBytes = capacityBytes;
            this.usedBytes = usedBytes;
        }
        
        public static Builder builder() {
            return new Builder();
        }
        
        public boolean isHealthy() { return healthy; }
        public long getCapacityBytes() { return capacityBytes; }
        public long getUsedBytes() { return usedBytes; }
        
        public static class Builder {
            private boolean healthy = false;
            private long capacityBytes = 0L;
            private long usedBytes = 0L;
            
            public Builder healthy(boolean healthy) {
                this.healthy = healthy;
                return this;
            }
            
            public Builder capacityBytes(long capacityBytes) {
                this.capacityBytes = capacityBytes;
                return this;
            }
            
            public Builder usedBytes(long usedBytes) {
                this.usedBytes = usedBytes;
                return this;
            }
            
            public DataServerHealthInfo build() {
                return new DataServerHealthInfo(healthy, capacityBytes, usedBytes);
            }
        }
    }
}
